package com.hkbigdata.flink.task;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.hkbigdata.flink.domain.MyCountMapFunction;
import com.hkbigdata.flink.domain.OrderProcess;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.util.Date;

import static com.hkbigdata.flink.utils.Constants.*;

public class C02_TotalOrder {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);
        // 如果是精准一次，必须开启checkpoint（后续章节介绍）
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);

        // 从Kafka读取数据
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
//                .setBootstrapServers("hadoop102:9093") // 指定kafka节点的地址和端口
                .setBootstrapServers(KAFKA_BOOSTRAP_SERVERS) // 指定kafka节点的地址和端口
                .setGroupId("order")  // 指定消费者组的id
                .setTopics(KAFKA_TOPIC_CDC)   // 指定消费的 Topic
                .setValueOnlyDeserializer(new SimpleStringSchema()) // 指定 反序列化器，这个是反序列化value
                .setStartingOffsets(OffsetsInitializer.earliest())  // flink消费kafka的策略
                .build();

        SingleOutputStreamOperator<String> orderDS = env
                .fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafkasource")
                .flatMap(new FlatMapFunction<String, OrderProcess>() {
                             @Override
                             public void flatMap(String value, Collector<OrderProcess> out) throws Exception {
                                 // System.out.println(value);
                                 try {
                                     ObjectMapper objectMapper = new ObjectMapper();
                                     JsonNode jsonNode = objectMapper.readTree(value);
                                     OrderProcess orderProcess = new OrderProcess();
                                     String table = jsonNode.get("source").get("table").asText();
                                     String operation = jsonNode.get("op").asText();
                                     long time = jsonNode.get("after").get("create_time").asLong();
                                     if ("ord_order".equals(table) && "c".equals(operation)) {
                                         orderProcess.setTable(table);
                                         orderProcess.setOperation(operation);
                                         orderProcess.setCreateTime(time);
                                         System.out.println("数据=" + orderProcess + ",recordTs=" + new Date(time));
                                         out.collect(orderProcess);
                                     }
                                 } catch (Exception e) {
                                     System.out.println("异常为：" + e.getMessage() + " 异常数据为：" + value);
                                 }
                             }
                         }
                ).map(new MyCountMapFunction()).process(new ProcessFunction<String, String>() {
                    @Override
                    public void processElement(String value, ProcessFunction<String, String>.Context ctx, Collector<String> out) throws Exception {
                        System.out.println("processElement=" + value);
                        out.collect(value);
                    }
                });

        orderDS.print();
        /**
         * Kafka Sink:
         * TODO 注意：如果要使用 精准一次 写入Kafka，需要满足以下条件，缺一不可
         * 1、开启checkpoint（后续介绍）
         * 2、设置事务前缀
         * 3、设置事务超时时间：   checkpoint间隔 <  事务超时时间  < max的15分钟
         */
//
//        KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
//                // 指定 kafka 的地址和端口，
//                .setBootstrapServers("192.168.64.102:9092")
//                // 指定序列化器：指定Topic名称、具体的序列化
//                .setRecordSerializer(
//                        KafkaRecordSerializationSchema.<String>builder()
//                                .setTopic("order_total")
//                                .setValueSerializationSchema(new SimpleStringSchema())
//                                .build()
//                )
//                // 写到kafka的一致性级别： 精准一次、至少一次
//                .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
//                // 如果是精准一次，必须设置 事务的前缀
//                .setTransactionalIdPrefix("task-process2-")
//                // 如果是精准一次，必须设置 事务超时时间: 大于checkpoint间隔，小于 max 15分钟
//                .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 10 * 60 * 1000 + "")
//                .build();
//        orderDS.sinkTo(kafkaSink);


        FlinkJedisPoolConfig.Builder builder = new FlinkJedisPoolConfig.Builder().setHost(REDIS_HOST).setPort(REDIS_PORT);
        FlinkJedisPoolConfig build = builder.build();
        orderDS.addSink(new RedisSink<>(build, new RedisMapper<String>() {
            @Override
            public RedisCommandDescription getCommandDescription() {
                return new RedisCommandDescription(RedisCommand.SET);
            }

            @Override
            public String getKeyFromData(String data) {
                return REDIS_KEY;
            }

            @Override
            public String getValueFromData(String data) {
                return data;
            }
        }));

        env.execute();

    }
}
